package com.cn.file;

import com.cn.exception.MyException;
import org.apache.commons.lang3.StringUtils;

import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.lang.reflect.Method;
import java.nio.ByteBuffer;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.security.AccessController;
import java.security.PrivilegedAction;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 底层存储
 */
public class MappFile extends ReferenceResource {

    /**
     * page cache 大小：4KB
     */
    public static final int OS_PAGE_SIZE = 1024 * 1;

    /**
     * 持续内存使用大小
     * JVM中映射的虚拟内存总大小
     */
    private static final AtomicLong TOTAL_MAPPED_VIRTUAL_MEMORY = new AtomicLong(0);

    /**
     * JVM中mmap的数量
     */
    private static final AtomicInteger TOTAL_MAPPED_FILES = new AtomicInteger(0);

    /**
     * 文件
     */
    private File file;

    /**
     * 映射的内存对象
     */
    private MappedByteBuffer mappedByteBuffer;

    private FileChannel fileChannel;

    /**
     * 当前文件的写位置
     */
    protected final AtomicInteger committedPosition = new AtomicInteger(0);
    /**
     * 上次刷盘的buffer位置
     */
    private final AtomicInteger flushedPosition = new AtomicInteger(0);

    /**
     * Message will put to here first, and then reput to FileChannel if writeBuffer is not null.
     */
    protected ByteBuffer writeBuffer = null;

    /**
     * 记录位移
     */
    private final AtomicInteger offset = new AtomicInteger(0);

    private int limit;

    /**
     * 文件写入的位置
     */
    protected final AtomicInteger wrotePosition = new AtomicInteger(0);

    protected final AtomicInteger commitPosition = new AtomicInteger(0);


    /**
     * 文件名称
     */
    private String fileName;
    /**
     * 文件的起始偏移量
     */
    private long fileFromOffset;

    private volatile long storeTimestamp = 0;
    private boolean firstCreateInQueue = false;
    private long fileSize;


    public MappFile(String path) {
        this.file = new File(path);
    }

    public MappFile(String path, long filesize) {
        init(path, filesize);
    }

    /**
     * 初始化
     * <p>
     * <p>
     * 1.如果是操作固定大小的文件，比如文件限制为 1M，那么指定filesize
     * 2.如果读取可变大小的文件，那么filesize设为0
     * </p>
     *
     * @param path     文件路径
     * @param filesize 文件大小
     */
    public void init(String path, long filesize) {
        if (StringUtils.isEmpty(path)) {
            throw new MyException("路径不能为空！");
        }
        this.fileSize = filesize;
        this.file = new File(path);
        this.fileFromOffset = Long.valueOf(this.file.getName());
        this.fileName = file.getName();

        ensureDirOK(this.file.getParent());

        boolean ok = false;
        try {
            fileChannel = new RandomAccessFile(path, "rw").getChannel();
            long size = filesize <= 0 ? this.fileChannel.size() : filesize;
            this.mappedByteBuffer = this.fileChannel.map(FileChannel.MapMode.READ_WRITE, 0, size);
            this.limit = this.mappedByteBuffer.limit();
            ok = true;
        } catch (Exception e) {

        } finally {
            if (!ok && this.fileChannel != null) {
                try {
                    this.fileChannel.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    /**
     * 新建文件目录
     *
     * @param dirName
     */
    public static void ensureDirOK(final String dirName) {
        if (dirName != null) {
            File f = new File(dirName);
            if (!f.exists()) {
                boolean result = f.mkdirs();
            }
        }
    }


    public AppendMessageResult doAppend(MessageStore msg, AppendMessageCallback cb) {
        int currentPos = this.getWrotePosition().get();
        if (currentPos < this.fileSize) {
            writeBuffer = writeBuffer == null ? this.mappedByteBuffer.slice() : writeBuffer;
            //每次写入的时候重置position
            writeBuffer.position(currentPos);
            AppendMessageResult appendMessageResult = cb.doAppend(this.getFileFromOffset(), writeBuffer, (int) this.fileSize - currentPos, msg);
            //修改可写入的position
            this.getWrotePosition().addAndGet(appendMessageResult.getWroteBytes());
            return appendMessageResult;
        }
        return new AppendMessageResult(AppendMessageStatus.UNKNOWN_ERROR, 0, 0, null, 0, 0, 0, 0);
    }

    public boolean appendMsg(byte[] data) {
        int position = this.wrotePosition.get();
        //保证不能超出
        if ((position + data.length) <= this.fileSize) {
            try {
                this.fileChannel.position(position);
                this.fileChannel.write(ByteBuffer.wrap(data));
                this.wrotePosition.addAndGet(data.length);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        return false;
    }

    public long appendMsg1(byte[] data) {
        int position = this.wrotePosition.get();
        //保证不能超出
        if ((position + data.length) <= this.fileSize) {
            try {
                this.fileChannel.position(position);
                this.fileChannel.write(ByteBuffer.wrap(data));
                return this.wrotePosition.addAndGet(data.length);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        return 0l;
    }

    /**
     * 读取
     * <p>
     * 假设1个文件大小总是是1000，现在我想从位置400的地方开始读取，那么先将position更改到400，然后计算lenth长度（调用 remaining 函数即可），一般data数组长度与length一致；
     * l    ength此时为600，必须保证length 小于等于 600
     * </p>
     *
     * @param data   -- 目标数组
     * @param offset -- 起始位置
     * @param length -- buff大小（可读取大小）
     * @return
     */
    public byte[] getBytes(byte[] data, int offset, int length) {
        this.mappedByteBuffer.get(data, offset, length);
        return data;
    }

    /**
     * 写入文件，默认从下标0开始
     *
     * @param data
     */
    public void write2File(byte[] data) {
        this.mappedByteBuffer.putInt(data.length);
        this.mappedByteBuffer.put(data);
    }

    public void write(ByteBuffer writeBuffer) throws IOException {
        this.fileChannel.write(writeBuffer);
    }

    /**
     * 是否已经写满
     *
     * @return
     */
    public boolean ifFull() {
        return this.fileSize == getReadPosition();
    }

    public boolean flush(int flushLeastPages) {
        //判断是否可以刷新
        if (this.isAble2Flush(flushLeastPages) && !this.ifFull()) {
            if (this.hold()) {
                int value = getReadPosition();
                try {
                    if (this.fileChannel.position() != 0) {
                        this.fileChannel.force(false);
                    } else {
                        this.mappedByteBuffer.force();
                    }
                    this.flushedPosition.set(value);
                    return true;
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    this.release();
                }
            }

        }
        return false;
    }

    /**
     * 是否可以刷盘
     *
     * @param flushLeastPages
     * @return
     */
    private boolean isAble2Flush(int flushLeastPages) {
        if (flushLeastPages == 0) {
            return true;
        }
        int read = getReadPosition();
        int flush = this.flushedPosition.get();

        /**
         * 假设 flushLeastPages=4，那么就是当buffer缓存页数超过4的时候，就需要刷盘
         * 公式： (当前写入的位置 / 每页大小) - (上次刷盘的位置  /每页大小)
         */
        if (flushLeastPages > 0) {
            return ((read / OS_PAGE_SIZE) - (flush / OS_PAGE_SIZE)) >= flushLeastPages;
        }
        return false;
    }

    /**
     * 获取buffer当前的写入位置
     *
     * @return
     */
    public int getReadPosition() {
        return this.wrotePosition.get();
    }

    public void setWrotePosition(long offset) {
        this.wrotePosition.set((int) offset);
    }

    /**
     * 关闭 并释放内存空间
     */
    public void close() {
        if (this.fileChannel != null) {
            try {
                this.fileChannel.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        cleanBuffer(this.mappedByteBuffer);
    }

    /**
     * 释放内存空间
     *
     * @param buffer
     */
    public void cleanBuffer(ByteBuffer buffer) {
        if (buffer == null || !buffer.isDirect() || buffer.capacity() == 0) {
            return;
        }
        AccessController.doPrivileged(new PrivilegedAction<Object>() {
            public Object run() {
                try {
                    Method getCleanerMethod = buffer.getClass().getMethod("cleaner", new Class[0]);
                    getCleanerMethod.setAccessible(true);
                    sun.misc.Cleaner cleaner = (sun.misc.Cleaner) getCleanerMethod.invoke(buffer, new Object[0]);
                    cleaner.clean();
                } catch (Exception e) {
                    e.printStackTrace();
                }
                return null;
            }
        });
    }

    public SelectMappedBufferResult selectMappedBuffer(int pos, int size) {
        int readPosition = getReadPosition();
        if (pos <= readPosition && pos >= 0) {
            if (this.hold()) {
                ByteBuffer byteBuffer = this.mappedByteBuffer.slice();
                byteBuffer.position(pos);
                ByteBuffer byteBuffer1 = byteBuffer.slice();
                //读取大小
                byteBuffer1.limit(size);
                return new SelectMappedBufferResult((int) this.fileFromOffset + pos, size, byteBuffer1, this);
            }
        }
        return null;
    }


    public SelectMappedBufferResult selectMappedBuffer(int pos) {
        //获取当前数据文件的写入位置（pos不能超过该值）
        int readPosition = getReadPosition();
        if (pos <= readPosition && pos >= 0) {
            if (this.hold()) {
                //必须确保可读下标在范围内
                ByteBuffer byteBuffer = this.mappedByteBuffer.slice();
                //设置缓存块的下标
                byteBuffer.position(pos);
                //设置本次要读取的数据块大小
                int size = readPosition - pos;
                //复制一个数据块
                ByteBuffer byteBufferNew = byteBuffer.slice();
                //设置大小
                byteBufferNew.limit(size);
                return new SelectMappedBufferResult((int) this.fileFromOffset + pos, size, byteBufferNew, this);
            }
        }
        return null;
    }

    public ByteBuffer sliceByteBuffer() {
        return this.mappedByteBuffer.slice();
    }

    /**
     * 恢复文件
     */
    public void recoverFile() {
        if (this.mappedByteBuffer != null) {
            ByteBuffer byteBuffer = this.mappedByteBuffer.slice();
            int t = 0;
            while (true) {
                int size = reloadFile(byteBuffer);
                if (size <= 0) {
                    break;
                }
                t = t + size;
            }
            //可以写入的位置
            this.wrotePosition.set(t);
            System.out.println(this.file.getPath() + "->当前position:" + this.getWrotePosition().get());
        }
    }

    /**
     * 恢复文件
     *
     * @param byteBuffer
     * @return
     */
    public int reloadFile(ByteBuffer byteBuffer) {

        // 1. 消息总长度
        int totalSize = byteBuffer.getInt();//4位TOTALSIZE
        if (totalSize <= 0) return 0;

        // 2. topic长度
        byte topicLength = byteBuffer.get();

        //3. topic内容
        byte[] topics = new byte[topicLength];
        byteBuffer.get(topics);
//        System.out.println(new java.lang.String(topics, MessageUtil.CHARSET_UTF8));

        //4. queueId
        int queueId = byteBuffer.getInt();
//        System.out.println("队列id:" + queueId);

        //5. 当前消息在queue队列中的个数
        long queueoffset = byteBuffer.getLong();


        //6. 消息在commitlog文件中的物理位置
        long physicaloffset = byteBuffer.getLong();
//        System.out.println("物理位置：" + physicaloffset);

        //7. 消息
        int bodyLength = byteBuffer.getInt();//4位
        if (bodyLength > 0) {
            byte[] datas = new byte[bodyLength];//totalsize
            byteBuffer.get(datas);//消息内容
//            System.out.println(new java.lang.String(datas, MessageUtil.CHARSET_UTF8));
        }

        //7. 存储时间
        long storeTimestamp = byteBuffer.getLong();
        //System.out.println(storeTimestamp);

        return 4 //存放4位totalsize
                + 1 //1位 存放 topic的长度
                + topicLength //topic的大小
                + 4 // queueId
                + 8 // queueoffset
                + 8 // 物理存储位置
                + 4//4位 存放body内容长度
                + bodyLength//body大小
                + 8 //存储时间
                ;
    }

    @Override
    public boolean cleanup(long currentRef) {
        return false;
    }

    public void clean(ByteBuffer byteBuffer) {
        if (byteBuffer == null || byteBuffer.isDirect() || byteBuffer.capacity() == 0) {
            return;
        }
    }

    public void setFileSize(long fileSize) {
        this.fileSize = fileSize;
    }

    public AtomicInteger getCommittedPosition() {
        return committedPosition;
    }


    public AtomicInteger getFlushedPosition() {
        return flushedPosition;
    }

    public AtomicInteger getOffset() {
        return offset;
    }


    public AtomicInteger getWrotePosition() {
        return wrotePosition;
    }

    public long getFileFromOffset() {
        return fileFromOffset;
    }

    public void setFileFromOffset(long fileFromOffset) {
        this.fileFromOffset = fileFromOffset;
    }

    public MappedByteBuffer getMappedByteBuffer() {
        return mappedByteBuffer;
    }

    public String getFileName() {
        return fileName;
    }

    public long getStoreTimestamp() {
        return storeTimestamp;
    }

    public FileChannel getFileChannel() {
        return fileChannel;
    }

    public long getFileSize() {
        return fileSize;
    }

    public AtomicInteger getCommitPosition() {
        return commitPosition;
    }

    public void setFlushedPosition(Long position) {
        this.flushedPosition.set(position.intValue());
    }

    public void setCommittedPosition(Long position) {
        this.committedPosition.set(position.intValue());
    }
}
